package com.bestcem.xm.member.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.bestcem.xm.common.core.domain.web.ServiceResult;
import com.bestcem.xm.common.core.sql.ColumnDesc;
import com.bestcem.xm.common.core.sql.CompileFeature;
import com.bestcem.xm.common.core.sql.SQLConditionCompiler;
import com.bestcem.xm.common.core.uitls.DateUtil;
import com.bestcem.xm.common.core.uitls.IdUtil;
import com.bestcem.xm.common.core.uitls.JsonUtil;
import com.bestcem.xm.componet.redis.service.RedisService;
import com.bestcem.xm.label.service.LabelMemberService;
import com.bestcem.xm.member.controller.param.group.CreateGroupParam;
import com.bestcem.xm.member.controller.param.group.UpdateGroupParam;
import com.bestcem.xm.member.controller.v2.param.CountGroupMemberParam;
import com.bestcem.xm.member.controller.vo.*;
import com.bestcem.xm.member.dao.GroupDao;
import com.bestcem.xm.member.dao.MemberDao;
import com.bestcem.xm.member.entity.mysql.MemberEntity;
import com.bestcem.xm.member.entity.pojo.GroupDO;
import com.bestcem.xm.member.entity.pojo.MemberDO;
import com.bestcem.xm.member.enums.MemberStatusEnum;
import com.bestcem.xm.member.enums.MemberTypeEnum;
import com.bestcem.xm.member.enums.NewMemberStatusEnum;
import com.bestcem.xm.member.grpc.service.param.imports.CheckImportParam;
import com.bestcem.xm.member.mq.message.GroupDeletedMessage;
import com.bestcem.xm.member.mq.message.GroupingMessage;
import com.bestcem.xm.member.mq.message.MemberCreationUpdMessage;
import com.bestcem.xm.member.service.GroupService;
import com.bestcem.xm.member.service.ImportService;
import com.bestcem.xm.member.service.MemberService;
import com.bestcem.xm.member.service.dto.MemberBatchDTO;
import com.bestcem.xm.member.service.mq.send.MemberMessageSendService;
import com.bestcem.xm.member.service.param.SelectMemberDynamicSql;
import com.bestcem.xm.member.service.param.UpdateMemberGroupDynamicSql;
import com.bestcem.xm.member.util.AutoIncrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.bestcem.xm.member.constant.Constants.*;

/**
 * 分组Service实现类
 *
 * @author ympan <yuming.pan@idiaoyan.com>
 * @version v1.0
 * @date 2021/11/24 16:35
 */
@Slf4j
@Service
public class GroupServiceImpl implements GroupService {
    private static final Logger EMAIL_LOG = LoggerFactory.getLogger("EmailErrorLog");

    @Resource
    private GroupDao groupDao;

    @Resource
    private AutoIncrUtil autoIncrUtil;

    @Resource
    private MemberDao memberDao;

    @Resource
    private MemberService memberService;

//    @Resource
//    private RabbitMessageSender rabbitMessageSender;

    //@Resource
    //private PackUtil packUtil;

    @Resource
    private RedisService redisService;

//    @Resource
//    private RabbitMessageSender mqMessageSender;

//    @Resource
//    RabbitMqConstantConfig rabbitMqConstantConfig;

    @Resource
    ImportService importService;

    @Resource
    private MemberMessageSendService memberMessageSendService;

    /**
     * 标签服务
     */
    @Resource
    private LabelMemberService labelMemberService;

    @Override
    public ServiceResult<CreateGroupVO> createGroup(CreateGroupParam param, String orgId) {
        // 根据公司标识获取当前套餐
        /*PackageDTO packageDTO = packUtil.getPackageByOrgId(orgId);
        if (Objects.isNull(packageDTO)) {
            log.error("【联系人】获取套餐失败，orgId:{}", orgId);
            return ServiceResult.fail("创建分组失败");
        }

        // 校验是否拥有分组权限
        boolean hasPermission = packUtil.hasMemberGroupQuataPermission(packageDTO);
        if (!hasPermission) {
            return ServiceResult.fail("没有分组权限");
        }

        // 校验 当前分组数量是否 < 分组限额
        Long groupCount = groupDao.countByOrgIdAndIsSample(orgId);
        if (log.isInfoEnabled()) {
            log.info("【联系人】公司{}的分组数量为{}(不包含示例数据)", orgId, groupCount);
        }
        Integer groupQuota = packageDTO.getGroupQuota();
        if (Objects.isNull(groupQuota)) {
            log.error("group quota is null, orgId:{}, packageId:{}", orgId, packageDTO.getPackageId());
            return ServiceResult.fail("分组限额为空");
        }
        if (groupQuota != -1 && groupCount >= groupQuota) {
            return ServiceResult.fail("超出分组限额");
        }
        if (log.isInfoEnabled()) {
            log.info("【联系人】公司{}的分组限额为{}", orgId, groupQuota);
        }*/

        // 校验名称是否重复
        Long nameCount = groupDao.countByOrgIdAndName(orgId, param.getName());
        if (nameCount != 0) {
            return ServiceResult.fail("分组名称已存在");
        }

        // 校验分组条件是否有效
        String condition = JsonUtil.tryConvertToJSONString(param.getCondition());
        int features = CompileFeature.of(CompileFeature.CHECK_LABEL_EXISTS,
                CompileFeature.CHECK_EXTEND_FIELD_EXISTS,
                CompileFeature.SKIP_SURVEY_CONDITION);
        ServiceResult<Wrapper<MemberEntity>> compileResult = labelMemberService.compileCondition(orgId, condition, features);
        if (!compileResult.isSuccess()) {
            log.error("【联系人】分组，条件解析失败，orgId:{} condition:{}", orgId, param.getCondition());
            return ServiceResult.fail(compileResult.getMsg());
        }

        // 创建分组对象
        GroupDO group = new GroupDO();
        group.setName(param.getName());

        group.setAuto(Objects.nonNull(param.getAuto()) && param.getAuto() ? 1 : 0);
        group.setCondition(JSONObject.toJSONString(param.getCondition()));
        String syncToken = IdUtil.uuid();
        group.setSyncToken(syncToken);
        group.setOrgId(orgId);
        group.setCreateTime(DateUtil.getCommonDate());
        group.setUpdateTime(DateUtil.getCommonDate());
        // 默认为非示例数据
        group.setIsSample(0);
        GroupDO groupDO = groupDao.save(group);

        // 发送同步消息
        ServiceResult<Void> serviceResult = sendGroupMessage(orgId, groupDO.getId(), syncToken, groupDO.getSeq());
        if (!serviceResult.isSuccess()) {
            return ServiceResult.fail(serviceResult.getMsg());
        }
        return ServiceResult.success(new CreateGroupVO(group.getId()));
    }


    @Override
    public ServiceResult<UpdateGroupVO> updateGroup(String id, UpdateGroupParam param, String orgId) {
        // 根据公司标识获取当前套餐
        /*PackageDTO packageDTO = packUtil.getPackageByOrgId(orgId);
        if (Objects.isNull(packageDTO)) {
            log.error("【联系人】获取套餐失败，orgId:{}", orgId);
            return ServiceResult.fail("更新分组失败");
        }

        // 校验是否拥有分组权限
        boolean hasPermission = packUtil.hasMemberGroupQuataPermission(packageDTO);
        if (!hasPermission) {
            return ServiceResult.fail("没有分组权限");
        }*/

        // 校验分组是否存在
        GroupDO existedGroup = groupDao.findById(id);
        if (Objects.isNull(existedGroup)) {
            return ServiceResult.fail("分组不存在");
        }
        Long seq = existedGroup.getSeq();

        // 校验名称是否重复
        long count = groupDao.countByOrgIdAndName(orgId, param.getName());
        if (count != 0 && !id.equals(existedGroup.getId())) {
            return ServiceResult.fail("分组名称已存在");
        }

        // 校验分组条件是否有效
        if (Objects.nonNull(param.getCondition())) {
            String condition = JsonUtil.tryConvertToJSONString(param.getCondition());
            int features = CompileFeature.of(CompileFeature.CHECK_LABEL_EXISTS,
                    CompileFeature.CHECK_EXTEND_FIELD_EXISTS,
                    CompileFeature.SKIP_SURVEY_CONDITION);
            ServiceResult<Wrapper<MemberEntity>> compileResult = labelMemberService.compileCondition(orgId, condition, features);
            if (!compileResult.isSuccess()) {
                log.error("【联系人】分组，条件解析失败，orgId:{} condition:{}", orgId, param.getCondition());
                return ServiceResult.fail(compileResult.getMsg());
            }
        }

        // 更新分组
        GroupDO group = new GroupDO();
        group.setId(id);
        group.setName(param.getName());
        if (Objects.nonNull(param.getCondition())) {
            group.setCondition(JSON.toJSONString(param.getCondition()));
        }
        group.setAuto(Objects.nonNull(param.getAuto()) && param.getAuto() ? 1 : 0);
        group.setUpdateTime(DateUtil.getCommonDate());
        String syncToken = null;
        if (Objects.nonNull(param.getCondition())) {
            syncToken = IdUtil.uuid();
            group.setSyncToken(syncToken);
        }
        groupDao.updateGroup(group);

        // 发送同步消息
        if (Objects.nonNull(param.getCondition())) {
            ServiceResult<Void> serviceResult = sendGroupMessage(orgId, id, syncToken, seq);
            if (!serviceResult.isSuccess()) {
                return ServiceResult.fail(serviceResult.getMsg());
            }
        }
        return ServiceResult.success(new UpdateGroupVO(id));
    }

    @Override
    public ServiceResult<ListGroupVO> listGroup(String orgId, String userId, Integer superAdmin, Integer seniorAdmin) {
        // 记录返回结果
        List<GroupVO> groupCounts = new ArrayList<>();

        // 查询分组
        List<GroupDO> allGroups = new ArrayList<>();
        List<GroupDO> groups = groupDao.findByOrgId(orgId);

        // 查询“全部”联系人
        GroupDO allGroup = new GroupDO();
        allGroup.setId("");
        allGroup.setName("全部");
        allGroups.add(allGroup);
        allGroups.addAll(groups);

        // 查询是否有未分组完成的联系人
        MemberDO existedMember = memberDao.findOneByOrgIdAndMemberTypeAndStatusAndGroupSyncToken(orgId,
                MemberTypeEnum.INDIVIDUAL.getIndex(),
                MemberStatusEnum.ACTIVATED.getIndex());

        // 查询当前企业是否导入完成
        ServiceResult<String> importResult = importService.checkImport(new CheckImportParam(orgId, MemberTypeEnum.INDIVIDUAL.getIndex()));
        if (!importResult.isSuccess()) {
            return ServiceResult.fail("分组查询失败");
        }
        boolean importing = StringUtils.isNotBlank(importResult.getData());

        CountGroupMemberParam param = new CountGroupMemberParam();
        param.setOrgId(orgId);
        param.setMemberType(MemberTypeEnum.INDIVIDUAL.getIndex());
        param.setUserId(userId);
        param.setSuperRole(superAdmin);
        param.setIsSenior(seniorAdmin);

        // 根据existedMember和importing设置grouping状态
        for (GroupDO group : allGroups) {
            param.setGroupId(group.getId());
            ServiceResult<Integer> countResult = memberService.countGroupMember(param);

            if (!countResult.isSuccess()) {
                if (log.isErrorEnabled()) {
                    log.error("【联系人】查询分组联系人数量异常:{}", countResult.getMsg());
                    return ServiceResult.fail(countResult.getMsg());
                }
            }
            Long count = countResult.getData().longValue();
            GroupVO groupVO = new GroupVO();
            groupVO.setId(group.getId());
            groupVO.setName(group.getName());
            groupVO.setMemberCount(count);
            groupVO.setIsSample(Objects.nonNull(group.getIsSample()) && group.getIsSample() != 0);
            if (StringUtils.isNotBlank(group.getCondition())) {
                groupVO.setCondition(JSON.parseObject(String.valueOf(group.getCondition())));
            }
            Integer auto = group.getAuto();
            groupVO.setAuto(Objects.nonNull(auto) && auto.equals(1));
            if (Objects.nonNull(existedMember) || importing) {
                groupVO.setGrouping(Objects.nonNull(auto) && auto.equals(1));
                if (log.isInfoEnabled() && Objects.nonNull(existedMember)) {
                    log.info("【联系人】group member未完成，orgId:{}，memberId:{}", orgId, existedMember.getId());
                }
            } else {
                groupVO.setGrouping(StringUtils.isNotBlank(group.getSyncToken()));
            }
            groupCounts.add(groupVO);
        }
        return ServiceResult.success(new ListGroupVO(groupCounts));
    }

    @Override
    public ServiceResult<DeleteGroupVO> deleteGroup(String id, String orgId) {
        // 根据公司标识获取当前套餐
        /*PackageDTO packageDTO = packUtil.getPackageByOrgId(orgId);
        if (Objects.isNull(packageDTO)) {
            log.error("【联系人】获取套餐失败，orgId:{}", orgId);
            return ServiceResult.fail("删除分组失败");
        }

        // 校验是否拥有分组权限
        boolean hasPermission = packUtil.hasMemberGroupQuataPermission(packageDTO);
        if (!hasPermission) {
            return ServiceResult.fail("没有分组权限");
        }*/

        // 根据id查询分组，校验分组是否存在以便获取seq
        GroupDO existedGroup = groupDao.findById(id);
        if (Objects.isNull(existedGroup)) {
            return ServiceResult.fail("分组不存在");
        }
        Long seq = existedGroup.getSeq();

        // 根据id删除分组
        groupDao.deleteById(id);

        // 异步更新联系人的groups
        boolean flag = sendGroupDeletedMessage(id, orgId, seq);
        if (!flag) {
            return ServiceResult.fail("删除分组失败");
        }

        return ServiceResult.success(new DeleteGroupVO(id));
    }

    @Override
    public ServiceResult<Void> sendMemberCreationUpdMessage(String orgId, String memberId, String groupSyncToken) {
        MemberCreationUpdMessage message = new MemberCreationUpdMessage();
        message.setOrgId(orgId);
        message.setMemberId(memberId);
        message.setGroupSyncToken(groupSyncToken);
//        try {
//            rabbitMessageSender.sendMsg(rabbitMqConstantConfig.getAppBpMemberCreationUpdExchange(),
//                    rabbitMqConstantConfig.getAppBpMemberCreationUpdRoutingKey(), "member.member_creation_upd", message);
//        } catch (JsonProcessingException e) {
//            log.error("publish member creation/update message failed, message: {}", message, e);
//            return ServiceResult.fail("发送联系人创建/更新消息失败");
//        }
        //stream mq /2021.05.12
        memberMessageSendService.memberCreationUpdSend(message);
        return ServiceResult.success();
    }

    @Override
    public ServiceResult<Void> sendGroupMessage(String orgId, String groupId, String uuid, Long seq) {
        GroupingMessage message = new GroupingMessage();
        message.setOrgId(orgId);
        message.setGroupId(groupId);
        message.setSyncToken(uuid);
        message.setSeq(seq);
//        try {
//            rabbitMessageSender.sendMsg(rabbitMqConstantConfig.getAppBpMemberGroupingExchange(),
//                    rabbitMqConstantConfig.getAppBpMemberGroupingRoutingKey(), "member.grouping", message);
//        } catch (JsonProcessingException e) {
//            log.error("publish group condition modification message failed, message: {}", message, e);
//            return ServiceResult.fail("发送分组规则修改消息失败");
//        }
        //stream mq /2022.05.12
        memberMessageSendService.groupConditionModSend(message);

        return ServiceResult.success();
    }

    @Override
    public ServiceResult<Void> sendGroupMessages(String orgId) {
        List<GroupDO> groups = groupDao.listByOrgIdAndAuto(orgId, 1);
        for (GroupDO group : groups) {
            String groupId = group.getId();
            Long seq = group.getSeq();
            String syncToken = IdUtil.uuid();
            groupDao.updateGroupSyncToken(groupId, syncToken, DateUtil.getCommonDate());
            ServiceResult<Void> result = sendGroupMessage(orgId, groupId, syncToken, seq);
            if (!result.isSuccess()) {
                return ServiceResult.fail(result.getMsg());
            }
        }
        return ServiceResult.success();
    }

    @Override
    public ServiceResult<Void> groupMember(MemberCreationUpdMessage memberCreationUpdMessage) {
        if (log.isInfoEnabled()) {
            log.info("【groupMember性能监控】groupMember 开始 timestamp: {}", System.currentTimeMillis());
        }
        String orgId = memberCreationUpdMessage.getOrgId();
        String memberId = memberCreationUpdMessage.getMemberId();
        String groupSyncToken = memberCreationUpdMessage.getGroupSyncToken();

        try {
            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】根据联系人标识获取联系人 开始 timestamp: {}", System.currentTimeMillis());
            }
            // 根据联系人标识获取联系人
            MemberDO member = memberDao.findOneByIdAndStatus(memberId, orgId,
                    MemberTypeEnum.INDIVIDUAL.getIndex(),
                    MemberStatusEnum.ACTIVATED.getIndex());
            if (Objects.isNull(member)) {
                log.error("member not found, memberId:{}", memberId);
                // 删除联系人令牌
                memberDao.updateMemberGroupSyncToken(memberId, "", groupSyncToken);
                return ServiceResult.success();
            }
            String groupsStr = member.getGroups();
            List<Long> existedSeq = JSONArray.parseArray(groupsStr, Long.class);
            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】根据联系人标识获取联系人 结束 timestamp: {}", System.currentTimeMillis());
            }

            // 校验令牌
            if (!StringUtils.equals(groupSyncToken, member.getGroupSyncToken())) {
                if (log.isInfoEnabled()) {
                    log.info("联系人令牌失效 memberId:{}", memberId);
                }
                return ServiceResult.success();
            }

            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】根据公司标识获取当前套餐 开始 timestamp: {}", System.currentTimeMillis());
            }
            // 根据公司标识获取当前套餐
            /*PackageDTO packageDTO = packUtil.getPackageByOrgId(orgId);
            if (Objects.isNull(packageDTO)) {
                EMAIL_LOG.error("【联系人】获取套餐失败，orgId:{}", orgId);
                Date delayDate = new Date(System.currentTimeMillis() + THIRTY_SECOND_UNIT_MILLIS);
//                mqMessageSender.sendDelayMsg(rabbitMqConstantConfig.getAppMemberDelayExchange(),
//                        rabbitMqConstantConfig.getAppMemberDelayRoutingKey(), MemberTopics.MEMBER_REGROUP_MEMBER_TOPIC,
//                        memberCreationUpdMessage, delayDate);
                memberMessageSendService.memberRegroupMemberSend(memberCreationUpdMessage, delayDate);
                return ServiceResult.fail("获取套餐失败，orgId:" + orgId);
            }
            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】根据公司标识获取当前套餐 结束 timestamp: {}", System.currentTimeMillis());
            }

            // 校验是否拥有分组权限
            boolean hasPermission = packUtil.hasMemberGroupQuataPermission(packageDTO);
            if (!hasPermission) {
                if (log.isInfoEnabled()) {
                    log.info("没有分组权限，orgId:{}", orgId);
                }
                // 删除联系人令牌
                memberDao.updateMemberGroupSyncToken(memberId, null, groupSyncToken);
                return ServiceResult.success();
            }*/

            long timestamp = System.currentTimeMillis();
            List<Long> memberGroups = new ArrayList<>();

            // 获取当前公司下的分组列表
            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】获取当前公司下的分组列表 开始 timestamp: {}", System.currentTimeMillis());
            }
            List<GroupDO> groups = groupDao.findByOrgId(orgId);
            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】获取当前公司下的分组列表 结束 timestamp: {}", System.currentTimeMillis());
            }

            // 获取当前公司下不可自动分组的分组列表
            List<GroupDO> nonAutoGroups = groups.stream().filter(group -> !(Objects.nonNull(group.getAuto()) && group.getAuto() == 1)).collect(Collectors.toList());
            if (Objects.nonNull(existedSeq)) {
                for (GroupDO nonAutoGroup : nonAutoGroups) {
                    for (Long seq : existedSeq) {
                        if (seq.equals(nonAutoGroup.getSeq())) {
                            memberGroups.add(seq);
                        }
                    }
                }
            }

            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】遍历分组为联系人分组 开始 timestamp: {}", System.currentTimeMillis());
            }
            // 获取当前公司下可自动分组的分组列表
            List<GroupDO> autoGroups = groups.stream().filter(group -> Objects.nonNull(group.getAuto()) && group.getAuto() == 1).collect(Collectors.toList());
            // 只包含一个元素，用来做id查询
            List<String> memberIds = new ArrayList<>();
            memberIds.add(memberId);
            for (GroupDO group : autoGroups) {
                SelectMemberDynamicSql dynamicSql = new SelectMemberDynamicSql();
                dynamicSql.setMemberIds(memberIds);
                dynamicSql.setOrgId(orgId);
                dynamicSql.setMemberType(MemberTypeEnum.INDIVIDUAL.getIndex());
                dynamicSql.setStatus(NewMemberStatusEnum.MEMBER_STATUS_ACTIVATED.getIndex());
                // 解析条件，拼装sql
                Map<String, ColumnDesc> columnMapping = new HashMap<>(8);
                columnMapping.put("mobile", ColumnDesc.newInstance("phone"));
                columnMapping.put("store", ColumnDesc.newInstance("store_id", ColumnDesc.JDBC_TYPE_BINARY));
                columnMapping.put("openid", ColumnDesc.newInstance("open_id"));
                columnMapping.put("appid", ColumnDesc.newInstance("app_id"));
                columnMapping.put("alipay_user_id", ColumnDesc.newInstance("ali_pay_user_id"));
                columnMapping.put("institut_id", ColumnDesc.newInstance("institute_id"));
                columnMapping.put("institut_name", ColumnDesc.newInstance("institute_name"));
                columnMapping.put("employeeID", ColumnDesc.newInstance("employee_id"));

                ServiceResult<Wrapper<MemberEntity>> result = SQLConditionCompiler.newInstance(columnMapping)
                        .compile(JsonUtil.tryConvertToJSONObject(group.getCondition()));
                if (!result.isSuccess()) {
                    EMAIL_LOG.error("【联系人】parse condition failure,groupId:{}", group.getId());
                    continue;
                }
                dynamicSql.setWrapper(result.getData());
                if (log.isInfoEnabled()) {
                    log.info("即将进入筛选的分组为{}", group);
                }
                List<MemberDO> members = memberDao.filterMembersByDynamicSql(dynamicSql);
                if (!CollectionUtils.isEmpty(members)) {
                    memberGroups.add(group.getSeq());
                }
            }
            // 更新联系人的groups（分组）字段
            Integer updateResult = memberDao.updateMemberGroups(JSONObject.toJSONString(memberGroups), memberId, groupSyncToken);
            if (updateResult == 0) {
                // 删除联系人令牌
                memberDao.updateMemberGroupSyncToken(memberId, null, groupSyncToken);
                return ServiceResult.success();
            }
            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】遍历分组为联系人分组 结束 timestamp: {}", System.currentTimeMillis());
            }

            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】补偿逻辑——分组被更新 开始 timestamp: {}", System.currentTimeMillis());
            }
            // 补偿逻辑——分组被更新
            GroupDO updatedGroup = groupDao.findOneByUpdateTimeGt(orgId, DateUtil.getCommonDate(new Date(timestamp)));
            if (Objects.nonNull(updatedGroup)) {
                if (log.isInfoEnabled()) {
                    log.info("【联系人】groupMember过程中分组{}被修改，该消息将重新入队，orgId{} memberId{}", updatedGroup.getId(), orgId, memberId);
                }
                Date delayDate = new Date(System.currentTimeMillis() + THIRTY_SECOND_UNIT_MILLIS);
//                mqMessageSender.sendDelayMsg(rabbitMqConstantConfig.getAppMemberDelayExchange(),
//                        rabbitMqConstantConfig.getAppMemberDelayRoutingKey(), MemberTopics.MEMBER_REGROUP_MEMBER_TOPIC,
//                        memberCreationUpdMessage, delayDate);
                memberMessageSendService.memberRegroupMemberSend(memberCreationUpdMessage, delayDate);
                return ServiceResult.success();
            }
            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】补偿逻辑——分组被更新 结束 timestamp: {}", System.currentTimeMillis());
            }

            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】补偿逻辑——分组被删除 开始 timestamp: {}", System.currentTimeMillis());
            }
            // 补偿逻辑——分组被删除
            List<String> oldGroupIds = groups.stream().map(GroupDO::getId).collect(Collectors.toList());
            List<GroupDO> currGroups = groupDao.findByOrgId(orgId);
            List<String> currGroupIds = currGroups.stream().map(GroupDO::getId).collect(Collectors.toList());
            currGroupIds.retainAll(oldGroupIds);
            if (currGroupIds.size() < oldGroupIds.size()) {
                if (log.isInfoEnabled()) {
                    log.info("【联系人】groupMember过程中分组被删除，该消息将重新入队，orgId{} memberId{}", orgId, memberId);
                }
                Date delayDate = new Date(System.currentTimeMillis() + THIRTY_SECOND_UNIT_MILLIS);
//                mqMessageSender.sendDelayMsg(rabbitMqConstantConfig.getAppMemberDelayExchange(),
//                        rabbitMqConstantConfig.getAppMemberDelayRoutingKey(), MemberTopics.MEMBER_REGROUP_MEMBER_TOPIC,
//                        memberCreationUpdMessage, delayDate);
                memberMessageSendService.memberRegroupMemberSend(memberCreationUpdMessage, delayDate);

                return ServiceResult.success();
            }
            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】补偿逻辑——分组被删除 结束 timestamp: {}", System.currentTimeMillis());
            }

            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】删除联系人令牌 开始 timestamp: {}", System.currentTimeMillis());
            }
            // 删除联系人令牌
            memberDao.updateMemberGroupSyncToken(memberId, null, groupSyncToken);
            if (log.isInfoEnabled()) {
                log.info("【groupMember性能监控】删除联系人令牌 结束 timestamp: {}", System.currentTimeMillis());
            }
        } catch (Exception e) {
//            EMAIL_LOG.error("【联系人】遍历联系人分组为联系人分组失败，memberId: {}", memberId, e);
            log.error("【联系人】遍历联系人分组为联系人分组失败，memberId: {}", memberId, e);
            // 删除联系人令牌
            memberDao.updateMemberGroupSyncToken(memberId, null, groupSyncToken);
            return ServiceResult.fail("遍历联系人分组为联系人分组失败，memberId:" + memberId);
        }
        if (log.isInfoEnabled()) {
            log.info("【groupMember性能监控】groupMember 结束 timestamp: {}", System.currentTimeMillis());
        }
        return ServiceResult.success();
    }

    @Override
    public ServiceResult<Void> groupMembers(GroupingMessage groupingMessage) {
        if (Objects.isNull(groupingMessage)) {
            return ServiceResult.success();
        }
        if (log.isInfoEnabled()) {
            log.info("【groupMembers性能监控】groupMembers 开始 timestamp: {}", System.currentTimeMillis());
        }

        String orgId = groupingMessage.getOrgId();
        String groupId = groupingMessage.getGroupId();
        String syncToken = groupingMessage.getSyncToken();
        Long seq = groupingMessage.getSeq();

        if (log.isInfoEnabled()) {
            log.info("【groupMembers性能监控】校验分组同步令牌（获取到锁前） 开始 timestamp: {}", System.currentTimeMillis());
        }
        // 校验分组同步令牌匹配（收到消息后）
        ServiceResult<Void> checkSyncTokenResult = checkSyncToken(groupId, syncToken, orgId, seq, false);
        if (!checkSyncTokenResult.isSuccess()) {
            return ServiceResult.success();
        }
        if (log.isInfoEnabled()) {
            log.info("【groupMembers性能监控】校验分组同步令牌（获取到锁前） 结束 timestamp: {}", System.currentTimeMillis());
        }

        if (log.isInfoEnabled()) {
            log.info("【groupMembers性能监控】校验分组是否存在 开始 timestamp: {}", System.currentTimeMillis());
        }
        // 校验分组是否存在，如果不存在，因为还没有开始分组，所以直接结束即可
        GroupDO group = groupDao.findById(groupId);
        if (Objects.isNull(group)) {
            log.error("group not found, groupId:{}", groupId);
            return ServiceResult.success();
        }
        if (log.isInfoEnabled()) {
            log.info("【groupMembers性能监控】校验分组是否存在 结束 timestamp: {}", System.currentTimeMillis());
        }

        // 校验分组规则是否存在
        Object condition = group.getCondition();
        if (Objects.isNull(condition)) {
            log.error("group document is incomplete, one auto field must be matching one condition field, groupId:{}", group.getId());
            // 删除分组令牌
            groupDao.updateSyncTokenByIdAndSyncToken(groupId, "", syncToken, DateUtil.getCommonDate());
            return ServiceResult.success();
        }

        // redisson针对分组seq上锁
        String lockKey = GROUP_LOCK_PREFIX + (seq % GROUP_LOCK_TOTAL_COUNT);
        RLock lock = this.redisService.getLock(lockKey);

        boolean res = false;
        try {
            // 尝试加锁，最多等待30秒，自动续期，超过5min-5s的时候告警
            res = lock.tryLock(30L, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error("【联系人】获取锁被中断，groupId:{}，errorMsg:{}", groupId, e.getMessage());
        }
        if (res) {
            long start = System.currentTimeMillis();
            try {
                if (log.isInfoEnabled()) {
                    log.info("【groupMembers性能监控】校验分组同步令牌（获取到锁后）开始 timestamp: {}", System.currentTimeMillis());
                }
                // 校验分组同步令牌匹配（获取到锁后）
                checkSyncTokenResult = checkSyncToken(groupId, syncToken, orgId, seq, false);
                if (!checkSyncTokenResult.isSuccess()) {
                    return ServiceResult.success();
                }
                if (log.isInfoEnabled()) {
                    log.info("【groupMembers性能监控】校验分组同步令牌（获取到锁后）结束 timestamp: {}", System.currentTimeMillis());
                }

                if (log.isInfoEnabled()) {
                    log.info("【groupMembers性能监控】联系人分批次 开始 timestamp: {}", System.currentTimeMillis());
                }
                // 联系人分批次
                List<MemberBatchDTO> memberBatches = batchOrgMember(orgId);
                if (CollectionUtils.isEmpty(memberBatches)) {
                    // 删除分组令牌
                    groupDao.updateSyncTokenByIdAndSyncToken(groupId, "", syncToken, DateUtil.getCommonDate());
                    return ServiceResult.success();
                }
                if (log.isInfoEnabled()) {
                    log.info("【groupMembers性能监控】联系人分批次 结束 timestamp: {}", System.currentTimeMillis());
                }

                if (log.isInfoEnabled()) {
                    log.info("【groupMembers性能监控】按批次清除联系人中分组标记并分组 开始 timestamp: {}", System.currentTimeMillis());
                }
                // 按批次清除联系人中分组标记并分组
                ServiceResult<Void> clearAndSyncCurrGroupResult = clearAndSyncCurrGroup(groupId, syncToken, seq, condition, memberBatches, orgId);
                if (!clearAndSyncCurrGroupResult.isSuccess()) {
                    // 删除分组令牌
                    groupDao.updateSyncTokenByIdAndSyncToken(groupId, "", syncToken, DateUtil.getCommonDate());
                    return ServiceResult.fail(clearAndSyncCurrGroupResult.getMsg());
                }
                if (log.isInfoEnabled()) {
                    log.info("【groupMembers性能监控】按批次清除联系人中分组标记并分组 结束 timestamp: {}", System.currentTimeMillis());
                }

                if (log.isInfoEnabled()) {
                    log.info("【groupMembers性能监控】删除分组令牌 开始 timestamp: {}", System.currentTimeMillis());
                }
                // 删除分组令牌
                groupDao.updateSyncTokenByIdAndSyncToken(groupId, "", syncToken, DateUtil.getCommonDate());
                if (log.isInfoEnabled()) {
                    log.info("【groupMembers性能监控】删除分组令牌 结束 timestamp: {}", System.currentTimeMillis());
                }
            } catch (Exception e) {
                EMAIL_LOG.error("【联系人】遍历联系人为其分组失败，groupId: {}，ErrorMsg: {}", groupId, e.getMessage());
                // 删除分组令牌
                groupDao.updateSyncTokenByIdAndSyncToken(groupId, "", syncToken, DateUtil.getCommonDate());
                return ServiceResult.fail("【联系人】遍历联系人为其分组失败");
            } finally {
                long executeTime = System.currentTimeMillis() - start;
                if (log.isInfoEnabled()) {
                    log.info("【联系人】遍历联系人为其分组，groupId:{}，实际执行时间:{}ms", groupId, executeTime);
                }
                if (executeTime > GROUP_LOCK_WARNING_MILLISECOND_VALUE) {
                    EMAIL_LOG.error("【联系人】遍历联系人为其分组，执行时间超过告警值，groupId:{},实际执行时间:{}ms", groupId, executeTime);
                }
                // 手动释放锁
                lock.unlock();
            }
        } else {
            if (log.isInfoEnabled()) {
                log.info("【联系人】获取锁失败，orgId:{} groupId: {} seq:{}", orgId, groupId, seq);
            }
            Date delayDate = new Date(System.currentTimeMillis() + THIRTY_SECOND_UNIT_MILLIS);
//            mqMessageSender.sendDelayMsg(rabbitMqConstantConfig.getAppMemberDelayExchange(),
//                    rabbitMqConstantConfig.getAppMemberDelayRoutingKey(), MemberTopics.MEMBER_REGROUP_MEMBERS_TOPIC,
//                    groupingMessage, delayDate);
            memberMessageSendService.memberRegroupMembersSend(groupingMessage, delayDate);
            return ServiceResult.fail("获取锁失败");
        }
        if (log.isInfoEnabled()) {
            log.info("【groupMembers性能监控】groupMembers 结束 timestamp: {}", System.currentTimeMillis());
        }
        return ServiceResult.success();
    }

    @Override
    public ServiceResult<Void> updateMemberGroups(GroupDeletedMessage groupDeletedMessage) {
        String orgId = groupDeletedMessage.getOrgId();
        String groupId = groupDeletedMessage.getGroupId();
        Long seq = groupDeletedMessage.getSeq();

        // 联系人分批次
        List<MemberBatchDTO> memberBatches = batchOrgMember(orgId, seq);
        if (CollectionUtils.isEmpty(memberBatches)) {
            return ServiceResult.success();
        }

        ServiceResult<Void> result = pullGroups(seq, memberBatches);
        if (!result.isSuccess()) {
            //  todo 要不要把里面的内容try起来，如果异常则重新入队
            log.error("【联系人】pullGroups失败，groupId:{}", groupId);
            return ServiceResult.fail(result.getMsg());
        }
        return ServiceResult.success();
    }

    /**
     * 删除联系人分组seq
     *
     * @param seq
     * @param memberBatches
     * @return
     */
    private ServiceResult<Void> pullGroups(Long seq, List<MemberBatchDTO> memberBatches) {
        int i = 0;
        for (MemberBatchDTO memberBatch : memberBatches) {
            if (log.isInfoEnabled()) {
                log.info("【联系人】pullGroups，共{}个批次，当前第{}个批次", memberBatches.size(), ++i);
            }
            String orgId = memberBatch.getOrgId();
            Date startTime = memberBatch.getStartTime();
            Date endTime = memberBatch.getEndTime();
            memberDao.deleteGroup(orgId, MemberTypeEnum.INDIVIDUAL.getIndex(), MemberStatusEnum.ACTIVATED.getIndex(), seq,
                    startTime, endTime);
        }
        return ServiceResult.success();
    }


    /**
     * 将公司下联系人分批，按照时间
     *
     * @param orgId 企业标识
     * @param seq   分组序号
     * @return 联系人分批次实体列表
     * @author ympan <yuming.pan@idiaoyan.com>
     * @date 20:56 2021/12/6
     */
    private List<MemberBatchDTO> batchOrgMember(String orgId, Long seq) {
        if (log.isInfoEnabled()) {
            log.info("【联系人】batch member start, orgId: {}, timestamp: {}", orgId, System.currentTimeMillis());
        }

        Date currTime = DateUtil.getCommonDate();

        long rawTotalMemberCount = memberDao.countByOrgIdAndMemberTypeAndStatusAndGroupsAndCreateTime(orgId,
                MemberTypeEnum.INDIVIDUAL.getIndex(),
                MemberStatusEnum.ACTIVATED.getIndex(),
                seq, currTime);

        int totalMemberCount = (int) rawTotalMemberCount;
        if (totalMemberCount == 0) {
            return new ArrayList<>();
        }

        // 分批准备
        int batchSize = 1000;
        int totalBatchCount = totalMemberCount / batchSize;
        int lastBatchSize = totalMemberCount - totalBatchCount * batchSize;
        List<MemberBatchDTO> memberBatches = new ArrayList<>(totalBatchCount + 1);
        Date endTime = currTime;

        // 分批处理
        for (int batchIndex = 0; batchIndex < totalBatchCount; batchIndex++) {
            MemberDO memberDO = memberDao.findOneByOrgIdAndMemberTypeAndStatusAndGroupsAndCreateTime(orgId,
                    MemberTypeEnum.INDIVIDUAL.getIndex(),
                    MemberStatusEnum.ACTIVATED.getIndex(),
                    seq, endTime, batchSize - 1L, 1);

            if (memberDO != null) {
                Date startTime = memberDO.getCreateTime();
                memberBatches.add(new MemberBatchDTO(orgId, startTime, endTime));
                endTime = startTime;
            }
        }
        if (lastBatchSize > 0) {
            memberBatches.add(new MemberBatchDTO(orgId, null, endTime));
        }
        if (log.isInfoEnabled()) {
            log.info("【联系人】batch member end, orgId: {}, memberCount: {}, batchCount: {}, timestamp: {}",
                    orgId, totalMemberCount, memberBatches.size(), System.currentTimeMillis());
        }
        return memberBatches;
    }

    /**
     * 将公司下联系人分批，按照时间
     *
     * @param orgId 企业标识
     * @return 联系人分批次实体列表
     * @author ympan <yuming.pan@idiaoyan.com>
     * @date 20:56 2021/12/6
     */
    private List<MemberBatchDTO> batchOrgMember(String orgId) {
        if (log.isInfoEnabled()) {
            log.info("【联系人】batch member start, orgId: {}, timestamp: {}", orgId, System.currentTimeMillis());
        }

        // 以当前服务器时间为界，分批联系人（通过时间逆序获取，减少干预 联系人创建/修改的那个调度 的次数）
        Date currTime = DateUtil.getCommonDate();

        long rawTotalMemberCount = memberDao.countByOrgIdAndMemberTypeAndStatusAndCreateTime(orgId,
                MemberTypeEnum.INDIVIDUAL.getIndex(),
                MemberStatusEnum.ACTIVATED.getIndex(),
                currTime);

        int totalMemberCount = (int) rawTotalMemberCount;
        if (totalMemberCount == 0) {
            return new ArrayList<>();
        }

        // 分批准备
        int batchSize = 1000;
        int totalBatchCount = totalMemberCount / batchSize;
        int lastBatchSize = totalMemberCount - totalBatchCount * batchSize;
        List<MemberBatchDTO> memberBatches = new ArrayList<>(totalBatchCount + 1);
        Date endTime = currTime;

        // 分批处理
        for (int batchIndex = 0; batchIndex < totalBatchCount; batchIndex++) {

            MemberDO member = memberDao.findOneByOrgIdAndMemberTypeAndStatusAndCreateTime(orgId,
                    MemberTypeEnum.INDIVIDUAL.getIndex(),
                    MemberStatusEnum.ACTIVATED.getIndex(),
                    endTime, batchSize - 1L, 1);

            if (member != null) {
                Date startTime = member.getCreateTime();
                memberBatches.add(new MemberBatchDTO(orgId, startTime, endTime));
                endTime = startTime;
            }
        }
        if (lastBatchSize > 0) {
            memberBatches.add(new MemberBatchDTO(orgId, null, endTime));
        }
        if (log.isInfoEnabled()) {
            log.info("【联系人】batch member end, orgId: {}, memberCount: {}, batchCount: {}, timestamp: {}",
                    orgId, totalMemberCount, memberBatches.size(), System.currentTimeMillis());
        }
        return memberBatches;
    }

    /**
     * 按批次清除联系人中分组标记并分组
     *
     * @param groupId       分组标识
     * @param syncToken     分组syncToken
     * @param seq           分组序号
     * @param condition     分组条件
     * @param memberBatches 批次序列
     * @author ympan <yuming.pan@idiaoyan.com>
     * @date 15:28 2021/12/15
     */
    private ServiceResult<Void> clearAndSyncCurrGroup(String groupId, String syncToken, Long seq, Object condition, List<MemberBatchDTO> memberBatches, String orgId) {
        int i = 0;
        for (MemberBatchDTO memberBatch : memberBatches) {
            // 清除联系人中的当前分组
            // 校验分组同步令牌匹配（每批次执行前）
            ServiceResult<Void> checkSyncTokenResult = checkSyncToken(groupId, syncToken, orgId, seq, true);
            if (!checkSyncTokenResult.isSuccess()) {
                return ServiceResult.success();
            }

            if (log.isInfoEnabled()) {
                log.info("【联系人】clearAndSyncCurrGroup，共{}个批次，当前第{}个批次", memberBatches.size(), ++i);
            }

            Date startTime = memberBatch.getStartTime();
            Date endTime = memberBatch.getEndTime();

            // 构造 query
            memberDao.deleteGroup(orgId,
                    MemberTypeEnum.INDIVIDUAL.getIndex(),
                    MemberStatusEnum.ACTIVATED.getIndex(),
                    seq, startTime, endTime);

            // 同步联系人中的当前分组
            // 解析条件，构造 query
            // 解析条件，拼装sql
            UpdateMemberGroupDynamicSql dynamicSql = new UpdateMemberGroupDynamicSql();
            dynamicSql.setOrgId(orgId);
            dynamicSql.setMemberType(MemberTypeEnum.INDIVIDUAL.getIndex());
            dynamicSql.setStatus(NewMemberStatusEnum.MEMBER_STATUS_ACTIVATED.getIndex());
            dynamicSql.setSeq(seq);
            dynamicSql.setStartTime(startTime);
            dynamicSql.setEndTime(endTime);
            Map<String, ColumnDesc> columnMapping = new HashMap<>(8);
            columnMapping.put("mobile", ColumnDesc.newInstance("phone"));
            columnMapping.put("store", ColumnDesc.newInstance("store_id", ColumnDesc.JDBC_TYPE_BINARY));
            columnMapping.put("openid", ColumnDesc.newInstance("open_id"));
            columnMapping.put("appid", ColumnDesc.newInstance("app_id"));
            columnMapping.put("alipay_user_id", ColumnDesc.newInstance("ali_pay_user_id"));
            columnMapping.put("institut_id", ColumnDesc.newInstance("institute_id"));
            columnMapping.put("institut_name", ColumnDesc.newInstance("institute_name"));
            columnMapping.put("employeeID", ColumnDesc.newInstance("employee_id"));

            ServiceResult<Wrapper<MemberEntity>> result = SQLConditionCompiler.newInstance(columnMapping)
                    .compile(JsonUtil.tryConvertToJSONObject(condition));
            dynamicSql.setWrapper(result.getData());
            memberDao.updateMembersByDynamicSql(dynamicSql);
        }

        // 补偿措施，如果分组被删除，则发送分组删除的邮件
        GroupDO groupDO = groupDao.findById(groupId);
        if (Objects.isNull(groupDO)) {
            log.info("group not found, groupId:{}", groupId);
            boolean flag = sendGroupDeletedMessage(groupId, orgId, seq);
            if (!flag) {
                log.error("发送分组删除消息失败， groupId: {}", groupId);
                return ServiceResult.fail("发送分组删除消息失败");
            }
        }
        return ServiceResult.success();
    }

    /**
     * 校验分组token是否有效
     *
     * @param groupId   分组id
     * @param syncToken 分组syncToken
     * @param orgId     公司id
     * @param seq       分组序号
     * @param grouping  是否处于分组过程中
     * @author ympan <yuming.pan@idiaoyan.com>
     * @date 15:26 2021/12/15
     */
    private ServiceResult<Void> checkSyncToken(String groupId, String syncToken, String orgId, Long seq, boolean grouping) {
        GroupDO groupDO = groupDao.findById(groupId);
        if (Objects.isNull(groupDO)) {
            log.info("group not found, groupId:{}", groupId);

            if (grouping) {
                boolean flag = sendGroupDeletedMessage(groupId, orgId, seq);
                if (!flag) {
                    log.error("发送分组删除消息失败， groupId: {}", groupId);
                    return ServiceResult.fail("发送分组删除消息失败");
                }
            }
            return ServiceResult.fail("group not found");
        }
        if (!StringUtils.equals(syncToken, groupDO.getSyncToken())) {
            if (log.isInfoEnabled()) {
                log.info("【联系人】synchronization token is invalid. groupId:{}, token:{}", groupId, syncToken);
            }
            return ServiceResult.fail("synchronization token is invalid.");
        }
        return ServiceResult.success();
    }

    private boolean sendGroupDeletedMessage(String groupId, String orgId, Long seq) {
        // 异步更新联系人的groups
        GroupDeletedMessage message = new GroupDeletedMessage(orgId, groupId, seq);
//        try {
//            rabbitMessageSender.sendMsg(rabbitMqConstantConfig.getAppBpMemberGroupDeletedExchange(),
//                    rabbitMqConstantConfig.getAppBpMemberGroupDeletedRoutingKey(), "member.group_deleted", message);
//        } catch (JsonProcessingException e) {
//            EMAIL_LOG.error("发送分组删除消息失败， message: {}", message, e);
//            return false;
//        }
        //stream mq /2022.05.12
        memberMessageSendService.groupDeletedSend(message);
        return true;
    }
}
    